package edu.vt.ece.rt.dstm;

import java.io.IOException;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

import org.jgroups.Message;

import edu.vt.ece.rt.quorum.Cluster;
import edu.vt.ece.rt.quorum.ClustersManager;
import edu.vt.ece.rt.quorum.Resource;
import edu.vt.ece.rt.quorum.messsages.types.JoinCluster;
import edu.vt.ece.rt.quorum.quorum.QuroumFailureException;
import edu.vt.ece.rt.quorum.quorum.ReadQuorum;
import edu.vt.ece.rt.quorum.quorum.WriteQuorum;
import edu.vt.ece.rt.quorum.util.AbortException;
import edu.vt.ece.rt.quorum.util.Names;
import edu.vt.ece.rt.quorum.util.ReplyObjectStore;
import edu.vt.ece.rt.util.Logger.Logger;


/**
 * Application Interface to DSTM
 * @author Sudhanshu
 */
public class TMProxy{
	ReadQuorum rQ;
	WriteQuorum wQ;
	AtomicInteger requestReconfig; 

	static TMProxy tp;

	public static TMProxy getTMProxy(){
		if(tp == null){
			tp = new TMProxy();
			tp.config();
		}
		return tp;
	}

	public static void start(){
		if(tp == null){
			tp = new TMProxy();
			tp.config();
			tp.waitForQuorums();
		}
	}

	private void waitForQuorums() {
		try {
			Thread.sleep((10000L)* (ClustersManager.baseClusterId));
			ClustersManager.getCluster(Names.base).send(new Message(null, new JoinCluster(true)));
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		try {
			synchronized(Resource.waitObj) {
				while (Resource.qrNodes !=  (Integer.getInteger(Names.nodes)-1)) {
					Resource.waitObj.wait();
				}
				//			System.out.println("Object wait over for "+id +" objects registered "+Resource.objectsRegistered);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}


	}

	/*
	 * Initial TMProxy Should be called after finding the rQ & wQ for node
	 */
	public void config(){
		Logger.debug("Initiating the TMProxy");
		//		ClustersManager.calculateDistance();
		try {
			Thread.sleep((10000L)* (ClustersManager.baseClusterId));
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		getTMProxy().wQ = WriteQuorum.getQuorum();
		getTMProxy().rQ = ReadQuorum.getQuorum();
		requestReconfig = new AtomicInteger(0);
		Logger.debug("TM Initiation Completed");
	}

	/**
	 * Returns the Object copy if exist in quorum, else returns null
	 */
	public final QObject read(String Id, Long tnx) throws AbortException {
		QObject ob = null;

		//		try {
		/* check in parent's cache */
		Context context = ContextDelegator.getContext(tnx);
		if ((ob=context.checkParentSet(Id)) != null) {
			return ob;
		}

		int tries = 0;
		// Might be happening due to packet drop, or object is locked
		//			while (ob == null) {
		boolean resdo = true;

		//		while (redo) {
		//			boolean timeout = false;
		//			
		try {
			ob = rQ.read(Id, tnx, true, tries);
		} catch (TimeoutException t) {
			t.printStackTrace();
			 System.out.println("Timeout for read request of txn "+tnx);
	            try {
					Process p = Runtime.getRuntime().exec("killall java");
				} catch (IOException e) {
					e.printStackTrace();
				}

		}

		//			if (!timeout) break;
		//		}

		//				tries++;
		//				if (ob == null) {
		//					Logger.debug("Retrying read WTF for "+tnx);
		//				}
		//			}
		if (ob!= null && ob.getOwner() != 0) {
			Logger.debug("Reading a locked object");
			ob.owner.set(0);
		}

		if (ob == null) {
			/* for now only current transaction during read */
			throw new AbortException(tnx);
		}

		//		} catch (QuroumFailureException e) {
		//			Logger.debug("Quorum exception for node "+ ClustersManager.baseClusterId+" !!!!");			
		//			reconfigCluster();
		//			return read(Id, tnx);		//Make Network failure  invisible to application
		//		} catch (Exception e) {
		//			e.printStackTrace();
		//		}

		return ob;
	}

	/**
	 * To re-create the rQ, wQ
	 */
	private void reconfigCluster() {
		if(requestReconfig.incrementAndGet() == 1)  // if you are first, reconfigure cluster
			this.config();		//Initiate again to get new clusters
		while(requestReconfig.get() !=0);			//wait until reconfiguration done		
	}

	/**
	 * Returns the Object copy if exist in quorum, else returns null
	 */
	public QObject write(String Id, Long tnx) throws AbortException {

		QObject ob = null;

		/* check in parent's cache */
		Context context = ContextDelegator.getContext(tnx);

		if ((ob=context.checkWriteSet(Id)) != null) {
			return ob;
		} else if ((ob=context.checkCreateSet(Id)) != null) {
			return ob;
		} else if ((ob=context.checkParentWriteSet(Id)) != null) {
			ob = ob.clone();
			context.addToWriteSet(ob);
			return ob;
		} else if ((ob=context.checkParentCreateSet(Id)) != null) {
			ob = ob.clone();
			context.addToCreateSet(ob);
			return ob;			
		}
		
		int tries = 0;
		try {
			ob = rQ.read(Id, tnx, false, tries);
		} catch (TimeoutException t) {
			t.printStackTrace();
			 System.out.println("Timeout for write request of txn "+tnx);
            try {
				Process p = Runtime.getRuntime().exec("killall java");
			} catch (IOException e) {
				e.printStackTrace();
			}

		}


		if (ob!= null && ob.getOwner() != 0) {
			Logger.debug("Reading a locked object");
			ob.owner.set(0);
		}

		if (ob == null) {
			//			System.out.println(Id +" is locked right now");
			throw new AbortException(tnx);
		}
		//			return ob;

		//		QObject ob = null;
		//		try {
		//			
		//			Context context = ContextDelegator.getContext(tnx);
		//			if ((ob=context.checkParentWriteSet(Id)) != null) {
		//				return ob;
		//			}
		//			
		//			int tries = 0;
		////			while (ob == null) {
		//				ob = wQ.write(Id, tnx, false, tries);
		////				tries++;
		////				if (ob == null) {
		////					Logger.debug("Retrying write WTF for "+tnx);
		////				}
		////
		////			}
		//			
		//			if (ob != null && ob.getOwner() != 0) {
		//				Logger.debug("Reading a locked object");
		//				ob.owner.set(0);
		//			}
		////			
		//			if (ob == null) {
		//				throw new Exception();
		//			}
		////		} catch (QuroumFailureException e) {
		////			Logger.debug("Quorum for write exception for node "+ ClustersManager.baseClusterId+" !!!!");
		////			reconfigCluster();
		////			return write(Id, tnx);		//Make Network failure  invisible to application
		////		} catch (Exception e) {
		////			e.printStackTrace();
		////		}
		return ob;
	}

	public boolean requestCommit(Long tnx) throws AbortException{
		Boolean b = false;
		int tries = 0;
		Context c = ContextDelegator.getContext(tnx);
		//		try {
		//			while ((b = 
		//		if (c.isRoot()) {
		Long start = System.currentTimeMillis();
		try {
			b = wQ.requestCommit(tnx, tries);
			Logger.debug("Got the request commit response as "+b);		
		
		} catch (TimeoutException t) {
			t.printStackTrace();
			 System.out.println("Timeout for request commit of txn "+tnx);
			 
	            try {
					Process p = Runtime.getRuntime().exec("killall java");
				} catch (IOException e) {
					e.printStackTrace();
				}

		}
		Long end = System.currentTimeMillis();
		Logger.debug("write quorum validation "+(end - start));
			
		//		}
		//		else {
		//			Long start = System.currentTimeMillis();
		//			b = rQ.requestCommit(tnx, tries);
		//			//		Thread.sleep(500L);
		//			Long end = System.currentTimeMillis();
		//			Logger.debug("read quorum validation "+(end - start));
		//		}
		//
		if (b == false) {
			if (c.isChild() && c.isAbortParent()) {
				if (!c.isChild()) {
					Logger.debug("WTF is happening, child should have abort parent set");
				}
				throw new AbortException(c.getParentContextId());
			} else {
				throw new AbortException(c.getContextId());
			}	
			//			if (c.isChild()) {
			//				throw new AbortException(c.getParentContextId());
			//			} else {
			//				throw new AbortException(c.getContextId());
			//			}	
		}
		//					){
		//				tries++;
		//				if (b == null) {
		//					Logger.debug("Retrying request commit WTF for "+tnx);
		//				}				
		//			}
		//				wQ.requestCommit(tnx);
		//		} catch (QuroumFailureException e) {
		//			reconfigCluster();
		//			return false;			//fail the transaction, TODO: Might be retried to request commit
		//		} catch (Exception e) {
		//			e.printStackTrace();
		//		}
		return b;
	}

	public Cluster getReadChannel(){
		return rQ.getReaders();
	}

	public Cluster getWriteChannel(){
		return wQ.getWriters();
	}

	/**Publishes the object to the quorum, depending on 
	 * Initialised or uninitialised objects
	 * @param Object
	 */
	public void publishObject(QObject obj) {
		wQ.addToQuorum(obj);		
	}

	public boolean requestCommitChild(Long tnx, int i) throws Exception {
		Boolean b = false;
		int tries = 0;
		Context c = ContextDelegator.getContext(tnx);
		//		try {
		//			while ((b = 
		//		if (i >= 0) {
		//			Long start = System.currentTimeMillis();
		//			b = wQ.requestCommit(tnx, tries);
		////			Thread.sleep(500L);
		//			Long end = System.currentTimeMillis();
		//			Logger.debug("write quorum validation "+(end - start));
		//	} else {
		Long start = System.currentTimeMillis();
		b = rQ.requestCommit(tnx, tries);
		Long end = System.currentTimeMillis();
		Logger.debug("read quorum validation "+(end - start));
		//		}			
		if (b == null) {
			throw new Exception();
		}
		//					){
		//				tries++;
		//				if (b == null) {
		//					Logger.debug("Retrying request commit WTF for "+tnx);
		//				}				
		//			}
		//				wQ.requestCommit(tnx);
		//		} catch (QuroumFailureException e) {
		//			reconfigCluster();
		//			return false;			//fail the transaction, TODO: Might be retried to request commit
		//		} catch (Exception e) {
		//			e.printStackTrace();
		//		}
		return b;
	}

}
